package rabbitmq.rpc;

import com.rabbitmq.client.*;
import rabbitmq.Connect;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class Client {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = Connect.createConnection();

        // 创立信道
        Channel channel = connection.createChannel();
        channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, false, false, false, null);
        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, false, false, false, null);

        // 发送信息
        String uuid = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
                .correlationId(uuid) // 唯一表示
                .replyTo(Constants.RPC_RESPONSE_QUEUE) // 请求后所相应的队列
                .build();
        String msg = "hello pic...";

        channel.basicPublish("", Constants.RPC_REQUEST_QUEUE, props, msg.getBytes());
        System.out.println("发送消息");

        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
        DefaultConsumer response = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("收到消息：" + msg);
                if (uuid.equals(properties.getCorrelationId())) {
                    // 符合
                    queue.offer(msg);
                }
            }
        };

        // 接收服务端相应的结果
        channel.basicConsume(Constants.RPC_RESPONSE_QUEUE, true, response);
        queue.take(); // 阻塞 不然会一直走下去
        System.out.println("[RPC]:响应的消息为：" + msg);

        // 释放资源
//        channel.close();
//        connection.close();
    }
}
